package org.qpcr.community.device.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.qpcr.community.device.entity.CalibrationCmdEntity;
import org.qpcr.community.device.enums.CalibrationCommandStatuEnums;
import org.qpcr.community.device.enums.CalibrationCommandTypeEnums;
import org.qpcr.community.device.web.calibration.MessageHeader;
import org.qpcr.community.device.web.calibration.ResponseBody;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;

/**
 * 校准响应监听服务
 * 专门负责监听和处理校准响应队列中的消息
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class CalibrationResponseListenerService {

    private final ReactiveMessageQueueService reactiveMessageQueueService;
    @Lazy
    @Resource
    private CalibrationCmdService calibrationCmdService;
    private final AtomicInteger processedResponseCount = new AtomicInteger(0);

    /**
     * 内存缓存：用于存储校准和获取初始值指令的响应数据
     * key: messageId, value: 响应数据列表
     */
    private final ConcurrentHashMap<String, List<String>> responseCache = new ConcurrentHashMap<>();

    /**
     * 启动校准响应队列监听
     */
    public void startCalibrationResponseListener() {
        log.info("开始启动校准响应队列监听...");

        String responseQueueName = "device/calibration/response";

        // 使用改进的背压控制方法监听队列
        reactiveMessageQueueService.listenQueueWithBackpressureControl(responseQueueName,
                Duration.ofMillis(500)) // 增加轮询间隔，减少背压
            .doOnNext(responseData -> {
                int count = processedResponseCount.incrementAndGet();
                log.info("收到校准响应数据 [{}]: {}", count, responseData);
                processCalibrationResponse(responseData);
            })
            .doOnError(error -> {
                log.error("处理校准响应数据时发生错误: {}", error.getMessage(), error);
            })
            .doOnComplete(() -> {
                log.info("校准响应队列监听已停止");
            })
            .subscribe(
                // onNext
                responseData -> {
                    // 消息处理逻辑已在doOnNext中实现
                },
                // onError
                error -> {
                    log.error("校准响应队列监听发生严重错误，将尝试重启: {}", error.getMessage(), error);
                    // 延迟重启监听
                    restartListenerWithDelay(responseQueueName, Duration.ofSeconds(5));
                },
                // onComplete
                () -> {
                    log.info("校准响应队列监听正常完成");
                }
            );
    }

    /**
     * 延迟重启监听器
     */
    private void restartListenerWithDelay(String queueName, Duration delay) {
        log.info("将在 {} 秒后重启队列监听: {}", delay.getSeconds(), queueName);

        Mono.delay(delay)
            .doOnSuccess(tick -> {
                log.info("重启队列监听: {}", queueName);
                startCalibrationResponseListener();
            })
            .subscribe();
    }

    /**
     * 处理校准响应数据
     */
    private void processCalibrationResponse(String responseData) {
        try {
            // 解析响应数据
            JSONObject response = JSON.parseObject(responseData);
            MessageHeader header = response.getObject("header", MessageHeader.class);
            ResponseBody body = response.getObject("body", ResponseBody.class);

            String messageId = header.getMessageId();
            String status = body.getStatus();
            String deviceId = body.getDeviceId();
            String commandType = body.getCommandType();

            log.info("处理校准响应: messageId={}, status={}, deviceId={}, commandType={}",
                     messageId, status, deviceId, commandType);

            // 检查是否为需要缓存的指令类型
            if (isCacheableCommand(commandType)) {
                handleCacheableResponse(messageId, responseData, commandType);
            } else {
                // 非缓存指令直接处理
                updateCalibrationCommand(messageId, responseData);
            }
        } catch (Exception e) {
            log.error("解析校准响应数据失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 判断是否为需要缓存的指令类型
     */
    private boolean isCacheableCommand(String commandType) {
        return CalibrationCommandTypeEnums.CALIBRATION.getType().equals(commandType) ||
               CalibrationCommandTypeEnums.GET_INITIAL_VALUE.getType().equals(commandType) ||
               CalibrationCommandTypeEnums.GET_CALIBRATION_BEFORE_PARAM.getType().equals(commandType);
    }

    /**
     * 处理需要缓存的响应数据
     */
    private void handleCacheableResponse(String messageId, String responseData, String commandType) {
        // 获取或创建缓存列表
        List<String> cachedResponses = responseCache.computeIfAbsent(messageId, k -> new ArrayList<>());

        // 添加当前响应到缓存
        synchronized (cachedResponses) {
            cachedResponses.add(responseData);
            log.info("缓存响应数据: messageId={}, commandType={}, 当前缓存数量={}",
                    messageId, commandType, cachedResponses.size());
        }

        // 检查是否达到合并条件
        int requiredCount = getRequiredResponseCount(commandType);
        if (cachedResponses.size() >= requiredCount) {
            // 达到合并条件，执行合并和更新
            mergeAndUpdateResponses(messageId, cachedResponses, commandType);
        } else {
            log.info("响应数据不足，继续等待: messageId={}, commandType={}, 当前={}, 需要={}",
                    messageId, commandType, cachedResponses.size(), requiredCount);
        }
    }

    /**
     * 获取指定指令类型需要的响应数量
     */
    private int getRequiredResponseCount(String commandType) {
        if (CalibrationCommandTypeEnums.CALIBRATION.getType().equals(commandType)) {
            return 5; // 校准需要5条数据
        } else if (CalibrationCommandTypeEnums.GET_INITIAL_VALUE.getType().equals(commandType) || CalibrationCommandTypeEnums.GET_CALIBRATION_BEFORE_PARAM.getType().equals(commandType)) {
            return 4; // 获取初始值需要4条数据
        }
        return 1; // 默认只需要1条
    }

    /**
     * 合并响应数据并更新校准指令
     */
    private void mergeAndUpdateResponses(String messageId, List<String> cachedResponses, String commandType) {
        try {
            // 合并响应数据
            String mergedResponse = mergeResponseData(cachedResponses, commandType);

            log.info("响应数据合并完成: messageId={}, commandType={}, 合并前数量={}",
                    messageId, commandType, cachedResponses.size());

            // 更新校准指令
            updateCalibrationCommand(messageId, mergedResponse);

            // 清理缓存
            responseCache.remove(messageId);
            log.info("已清理缓存: messageId={}", messageId);

        } catch (Exception e) {
            log.error("合并响应数据失败: messageId={}, error={}", messageId, e.getMessage(), e);
        }
    }

    /**
     * 合并响应数据
     */
    private String mergeResponseData(List<String> cachedResponses, String commandType) {
        if (cachedResponses.isEmpty()) {
            throw new IllegalArgumentException("缓存响应列表为空");
        }

        // 使用第一条响应作为基础模板
        JSONObject baseResponse = JSON.parseObject(cachedResponses.get(0));
        JSONObject mergedResult = new JSONObject();

        // 根据指令类型进行不同的合并逻辑
        if (CalibrationCommandTypeEnums.CALIBRATION.getType().equals(commandType)) {
            mergedResult = mergeCalibrationData(cachedResponses);
        } else if (CalibrationCommandTypeEnums.GET_INITIAL_VALUE.getType().equals(commandType)) {
            mergedResult = mergeInitialValueData(cachedResponses);
        } else if (CalibrationCommandTypeEnums.GET_CALIBRATION_BEFORE_PARAM.getType().equals(commandType)) {
            mergedResult = mergeCalibrationBeforeParamData(cachedResponses);
        }

        // 更新响应体中的结果
        JSONObject body = baseResponse.getJSONObject("body");
        body.put("result", mergedResult);

        // 添加合并信息到响应中
//        body.put("mergedCount", cachedResponses.size());
//        body.put("mergedAt", System.currentTimeMillis());

        return baseResponse.toJSONString();
    }

    /**
     * 合并校准数据
     */
    private JSONObject mergeCalibrationData(List<String> cachedResponses) {
        JSONObject mergedResult = new JSONObject();

        StringBuilder calibrationParam = new StringBuilder();

        for (String responseData : cachedResponses) {
            try {
                JSONObject response = JSON.parseObject(responseData);
                JSONObject result = response.getJSONObject("body").getJSONObject("result");

                if (result != null) {

                    if (result.containsKey("calibrationParam")){
                        String resultString = result.getString("calibrationParam");
                        if (resultString.endsWith("\r\n")){
                            calibrationParam.append(resultString);
                        }else {
                            calibrationParam.append(resultString).append("\r\n");
                        }
                    }else {
                        mergedResult.putAll(result);
                    }
                }
            } catch (Exception e) {
                log.warn("解析校准响应数据失败: {}", e.getMessage());
            }
        }

        mergedResult.put("calibrationParam", calibrationParam.toString());
        return mergedResult;
    }

    private JSONObject mergeCalibrationBeforeParamData(List<String> cachedResponses) {
        JSONObject mergedResult = new JSONObject();

        StringBuilder calibrationParam = new StringBuilder();

        for (String responseData : cachedResponses) {
            try {
                JSONObject response = JSON.parseObject(responseData);
                JSONObject result = response.getJSONObject("body").getJSONObject("result");
                if (result != null) {
                    if (result.containsKey("calibrationParamBefore")){
                        String resultString = result.getString("calibrationParamBefore");
                        if (resultString.endsWith("\r\n")){
                            calibrationParam.append(resultString);
                        }else {
                            calibrationParam.append(resultString).append("\r\n");
                        }
                    }
                }
            } catch (Exception e) {
                log.warn("解析校准前参数响应数据失败: {}", e.getMessage());
            }
        }

        mergedResult.put("calibrationParamBefore", calibrationParam.toString());
        return mergedResult;
    }

    /**
     * 合并获取初始值数据
     */
    private JSONObject mergeInitialValueData(List<String> cachedResponses) {

        JSONObject mergedResult = new JSONObject();

        StringBuilder calibrationParam = new StringBuilder();

        for (String responseData : cachedResponses) {
            try {
                JSONObject response = JSON.parseObject(responseData);
                JSONObject result = response.getJSONObject("body").getJSONObject("result");
                if (result != null) {
                    if (result.containsKey("calibrationParam")){
                        String resultString = result.getString("calibrationParam");
                        if (resultString.endsWith("\r\n")){
                            calibrationParam.append(resultString);
                        }else {
                            calibrationParam.append(resultString).append("\r\n");
                        }
                    }
                }
            } catch (Exception e) {
                log.warn("解析获取初始值响应数据失败: {}", e.getMessage());
            }
        }

        mergedResult.put("calibrationParam", calibrationParam.toString());
        return mergedResult;
    }

    /**
     * 计算平均值
     */
    private double calculateAverage(List<Double> values) {
        if (values.isEmpty()) {
            return 0.0;
        }
        return values.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
    }

    /**
     * 更新校准指令 - 保持原本的更新逻辑
     */
    private void updateCalibrationCommand(String messageId, String responseData) {
        if (messageId != null) {
            calibrationCmdService.getByMessageId(messageId)
                .subscribe(
                    entity -> {
                        // 解析响应数据获取状态
                        JSONObject response = JSON.parseObject(responseData);
                        ResponseBody body = response.getObject("body", ResponseBody.class);
                        String status = body.getStatus();

                        // 更新状态和响应内容
                        entity.setStatus(status);
                        entity.setResponseBody(responseData);

                        // 使用原本的更新逻辑
                        calibrationCmdService.createUpdate()
                            .is(CalibrationCmdEntity::getId, entity.getId())
                            .set(CalibrationCmdEntity::getStatus, entity.getStatus())
                            .set(CalibrationCmdEntity::getResponseBody, entity.getResponseBody())
                            .set(CalibrationCmdEntity::getModifyTime,System.currentTimeMillis())
                            .execute()
                            .subscribe(
                                result -> log.info("校准指令状态已更新: messageId={}, status={}",
                                        messageId, entity.getStatus()),
                                error -> log.error("更新校准指令状态失败: messageId={}, error={}",
                                        messageId, error.getMessage())
                            );
                    },
                    error -> log.error("获取校准指令失败: messageId={}, error={}",
                            messageId, error.getMessage())
                );
        }
    }

    /**
     * 获取已处理的响应数量
     */
    public int getProcessedResponseCount() {
        return processedResponseCount.get();
    }

    /**
     * 重置响应处理计数器
     */
    public void resetProcessedResponseCount() {
        processedResponseCount.set(0);
        log.info("校准响应处理计数器已重置");
    }

    /**
     * 获取当前缓存状态
     */
    public Map<String, Integer> getCacheStatus() {
        Map<String, Integer> status = new ConcurrentHashMap<>();
        responseCache.forEach((messageId, responses) ->
            status.put(messageId, responses.size()));
        return status;
    }

    /**
     * 清理指定messageId的缓存
     */
    public void clearCache(String messageId) {
        responseCache.remove(messageId);
        log.info("已清理缓存: messageId={}", messageId);
    }

    /**
     * 清理所有缓存
     */
    public void clearAllCache() {
        int size = responseCache.size();
        responseCache.clear();
        log.info("已清理所有缓存，共清理 {} 个缓存项", size);
    }

    /**
     * 停止监听
     */
    public void stopListener() {
        log.info("停止校准响应队列监听");
        // 清理所有缓存
        clearAllCache();
    }
}
